package org.tigr.htc.server;

/*
This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation; either
version 2.1 of the License, or (at your option) any later version.

This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;

import org.apache.log4j.Logger;
import org.tigr.htc.common.ConfigUpdateLI;
import org.tigr.htc.common.HTCConfig;

public class RunnerMonitor implements Runnable, ConfigUpdateLI {
	static Logger log = Logger.getLogger(RunnerMonitor.class);

	// commands to monitor
	private static LinkedList<IRunner> runners = new LinkedList<IRunner> ();

	private static int PERIOD = Integer.parseInt(HTCConfig.getProperty("MONITOR_PERIOD"));

	public static int COUNT = 0;

	protected IRunner currentRunner;
	protected int ID;
    protected static boolean monitor = true;
    /**
     * The variable <code>monitors</code> holds the list of active monitors spawned to monitor
     * the jobs on the grid
     */
    protected static List<Thread> monitors = new ArrayList<Thread> ();

	public RunnerMonitor() { }

	/**
	 * The method <code>register</code> registers a runner that needs to be monitored for job status
	 * 
	 * @param p_runner a <code>IRunner</code> representing the runner that can be used to monitor the
     * job encapsulated by the runner.
	 */
	public static void register(IRunner p_runner) {
		log.debug("Add to monitor runner now monitoring #" + runners.size()
				+ " commands");
		synchronized (runners) {
			runners.add(p_runner);
			log.debug(" now notify ");
			log.debug(" now monitoring " + runners.size());
			runners.notifyAll();
		}
	}

	/**
	 * The method <code>spawnMonitor</code> spawns a monitor thread to monitor active runners.
     * 
	 */
	public synchronized static void spawnMonitor() {
		RunnerMonitor mon = new RunnerMonitor();
		mon.ID = RunnerMonitor.COUNT++;
		Thread monitor = new Thread(mon, " Monitor " + mon.ID);
        monitors.add(monitor); // Add this monitor to the list of monitors
		log.info(" start Monitor");
		monitor.start();
		monitor.setPriority(Thread.NORM_PRIORITY - 1);
		log.info(" monitor started ");
	}

	/**
     * The monitor attempts to remove the first runner from the waiting queue and checks to see if the
     * monitoring is complete, if not it adds the runner back to the queue to continue monitoring.
     * 
	 */
	public void run() {
		log.debug("Monitor loop starting...");

		HTCConfig.addConfigUpdateListener(this);

		// busy wait should change to some kind of selector
		while (monitor) {
			// check log file
			try {
                // Wait while there are no runners, if a runner is found then remove the runner
                // from the queue and start monitoring the runner
				synchronized (runners) {
					while (runners.size() < 1) {
						log.debug(" no runners to monitor going to sleep");
						runners.wait();
						log.debug("Wake and check monitor queue");
					}
                    
                    // If this thread is awakened because the monitor thread was stopped then
                    // stop
                    if (!monitor) {
                        if (log.isDebugEnabled()) {
                            log.debug(" stop monitoring.");
                            continue;
                        }
                    }
					log.debug(" removing first runner from queue");
					currentRunner = (IRunner) runners.removeFirst();
				}

				if (currentRunner != null) {
					Long id = new Long(currentRunner.getICommand().getID());
					Thread.currentThread().setName("(" + ID + ") Monitor Command " + id);
					log.debug("Start monitoring command id " + id);
                    
                    // Monitor the runner, if the monitoring is complete because the job finished
                    // then do nothing, otherwise add the runner back to the monitoring queue
					if (currentRunner.isRunnerComplete()) { // dequeue
						log.info("Removing runner id " + id + " from monitor queue ");
						currentRunner = null;
					} else { // put back in queue
						synchronized (runners) {
							runners.add(currentRunner);
						}
					}
					// Add a debug statement 
					log.debug("Finished monitoring command id " + id);					
				}
				
				Thread.currentThread().setName("Generic monitoring thread - " + ID);
				Thread.sleep(PERIOD);
			} catch (InterruptedException e) {
				log.fatal(" scheduler was interrupted ");
				log.fatal("...monitor loop stopping?");
				HTCConfig.removeConfigUpdateListener(this);
				
				// As this monitor thread is dying replace it with another thread
	            RunnerMonitor.spawnMonitor();
				return;
			} catch (Exception e) {
				log.fatal(" scheduler caught an exception ", e);
				log.fatal("...monitor loop stopping?");
				HTCConfig.removeConfigUpdateListener(this);				
				
				// As this monitor thread is dying replace it with another thread
	            RunnerMonitor.spawnMonitor();
				return;
			}
		}
	}

    /**
     * The method <code>stopMonitoring</code> signals all the monitoring threads to stop monitoring
     * in the next attempt to monitor.
     * 
     */
    public static void stopMonitoring () {
        if (log.isDebugEnabled()) {
            log.debug(" request to stop monitoring.");
        }
        synchronized (runners) {
            monitors.clear();
            runners.clear();
            monitor = false;
            runners.notifyAll(); // Send a notification to all the waiting threads to stop monitoring
        }
    }
    
    /**
     * The method <code>startMonitoring</code> signals all the monitor threads to resume monitoring
     */
    public static void startMonitoring () {
        if (log.isDebugEnabled()) {
            log.debug(" request to start monitoring.");
        }
        synchronized (runners) {
            monitor = true;
            runners.notifyAll(); // Send a notification to all the waiting threads to stop monitoring
        }
    }
    
	public void configUpdated() {
		log.debug("RunnerMonitor got a new config update... Reloading ");
		PERIOD = Integer.parseInt(HTCConfig.getProperty("MONITOR_PERIOD"));
	}

}
